/*================================================================
 *   Copyright (C) 2021 All rights reserved.
 *   
 *   文件名称：iomanager.cc
 *   创 建 者：汉森伯逸
 *   创建日期：2021/4/2
 *   描    述：
 *
 *================================================================*/
#include "iomanager.h"
#include <cstring>
#include <sys/fcntl.h>
#include <unistd.h>
#include "fdmanager.h"
#include "util/util.h"
#include "util/ptr_ex.h"
#include "mutex.h"

static const int EPOLL_EVENT_COUNT = 1024 * 20;

namespace hsby {

int IoWorker::FdContext::DelEvent(uint32_t ev) {
//    AssertEventFlag(ev);
    int res = 0;
    if (ev & EPOLLIN) {
        read_task.reset();
        res++;
    }
    if (ev & EPOLLOUT) {
        write_task.reset();
        res++;
    }
    event_flags &= ~ev;
    return res;
}

/**
 * @brief 主动触发事件
 */
int IoWorker::FdContext::TriggerEvent(uint32_t evs) {
    int res = 0;
    Task::Ptr task;
    if (evs & EPOLLIN && read_task) {
        IoManager::GetCurrentIoManager()->AddTask(read_task);
        DelEvent(EPOLLIN);
        res++;
    }
    if (evs & EPOLLOUT && write_task) {
        IoManager::GetCurrentIoManager()->AddTask(write_task);
        DelEvent(EPOLLOUT);
        res++;
    }
//    HSBY_LOG_SYSTEM_DEBUG << HSBY_FUNC_WITH_ARG_EXP(evs)
//                          << ", Complete trigger event, "
//                          << HSBY_VARS_EXP(event_flags);
    return res;
}

void IoWorker::FdContext::AssertEventFlag(uint32_t ev) {
    HSBY_ASSERT(ev);
    HSBY_ASSERT(!((ev & EPOLLIN) && (ev & EPOLLOUT)) );
}

IoWorker::IoWorker(IoManager* iom) : iom_(iom) {
    epoll_fd_ = epoll_create(1024 * 20);
    if (epoll_fd_ < 0) {
        HSBY_LOG_SYSTEM_ERROR << "epoll_create error, " << HSBY_ERROR_EXP;
        exit(-1);
    }

    // 初始化tickle管道
    if (pipe2(tickle_pipe_, O_NONBLOCK) != 0) {
        HSBY_LOG_SYSTEM_ERROR << "pipe(tickle_pipe_) error, " << HSBY_ERROR_EXP;
        exit(-1);
    }
    epoll_event epev;
    MemSetZero(epev);
    epev.events = EPOLLIN | EPOLLET;
    epev.data.fd = tickle_pipe_[0];
    int ret = epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, tickle_pipe_[0], &epev);
    if (ret != 0) {
        HSBY_LOG_SYSTEM_ERROR << "epoll_ctl error, " << HSBY_ERROR_EXP;
        exit(-1);
    }
}

IoWorker::~IoWorker() {
    //HSBY_LOG_SYSTEM_DEBUG << "On IoWorker::~IoWorker()";
    close(epoll_fd_);
    close(tickle_pipe_[0]);
    close(tickle_pipe_[1]);
}

void IoWorker::OnWorking() {
    epoll_event* epevents = new epoll_event[EPOLL_EVENT_COUNT];
    std::shared_ptr<epoll_event> sharedPtr_events(epevents, [](epoll_event* ptr){
        delete[] ptr;
    });

    TimeInterval ti;
    while(/*!IsStopped()*/ true ) {
        uint64_t next_time = GetNextTimerDistNow();
        int rt = 0;
        do {
            static const int MAX_TIMEOUT = 1000;
            if (next_time != ~0ull) {
                next_time = (int)(next_time > MAX_TIMEOUT) ? MAX_TIMEOUT : next_time;
            }else {
                next_time = MAX_TIMEOUT;
            }
            on_epoll_wait = true;
            rt = epoll_wait(epoll_fd_, epevents, EPOLL_EVENT_COUNT, next_time);
            HSBY_LOG_SYSTEM_DEBUG << "EpollWait" << ti;
            on_epoll_wait = false;
            HSBY_LOG_SYSTEM_DEBUG << "epoll_wait, rt=" << rt;

            // 非系统截断
            if (rt >= 0 || errno != EINTR) {
                break;
            }
        }while(true);

        // 处理定时器事件
        std::vector<std::function<void()>> cbs;
        ListExpriredCb(cbs);
        iom_->AddTask(cbs.begin(), cbs.end());

        // 处理事件
        for (int i = 0; i < rt; ++i) {
            epoll_event& epev = epevents[i];

	        // 处理Tickle
	        if (epev.data.fd == tickle_pipe_[0]) {
                char dummy[256];
                // 边缘触发，所以要一次性将数据读取出来
                while(read(epev.data.fd, dummy, sizeof(dummy)) > 0);
                continue;
            }

            // 处理epoll事件
            FdContext* fd_ctx = (FdContext*)epev.data.ptr;
	        HSBY_ASSERT(fd_ctx);
            // 如果出错或者挂起，设置重试
            if (epev.events & (EPOLLERR | EPOLLHUP)) {
                epev.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->event_flags;
            }

            // 获取返回的事件
            uint32_t ret_ev = epev.events & (EPOLLIN | EPOLLOUT);

            // 如果事件不存在
            if ((fd_ctx->event_flags & ret_ev) == 0) {
                continue;
            }
            // 触发事件
            TriggerEvent(fd_ctx->fd, ret_ev);
        }

        // 切回工作协程
        Fiber::GetCurrentFiber()->SwapOut();
    }
}

void IoWorker::Tickle() {
    if (on_epoll_wait) {
        write(tickle_pipe_[1], "T", 1);
    }
}

int IoWorker::DelEvent(size_t fd, uint32_t evs) {
    auto fd_ctx = GetFdContext(fd);
    if (!fd_ctx) {
        return 0;
    }

    if (!(evs & fd_ctx->event_flags)) {
        return 0;
    }

    // 操作epoll
    uint32_t left_ev_flags = (uint32_t)(fd_ctx->event_flags & ~evs);
    int op = left_ev_flags ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = left_ev_flags | EPOLLET;
    epevent.data.ptr = fd_ctx.get();
    if (epoll_ctl(epoll_fd_, op, fd, &epevent)) {
        return 0;
    }

    // 删除事件
    fd_ctx->event_flags = left_ev_flags;

    return 1;
}

int IoWorker::TriggerEvent(size_t fd, uint32_t evs) {
    auto fd_ctx = GetFdContext(fd);
    if (!fd_ctx) {
        return 0;
    }

    if (!(fd_ctx->event_flags & evs)) {
        return 0;
    }

    // 操作epoll
    uint32_t left_evs = (uint32_t)(fd_ctx->event_flags & ~evs);
    int op = left_evs ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = left_evs | EPOLLET;
    epevent.data.ptr = fd_ctx.get();
    if (epoll_ctl(epoll_fd_, op, fd, &epevent)) {
        return 0;
    }

    // 设置套接字标志
    return fd_ctx->TriggerEvent(evs);
}

int IoWorker::CancelAll(size_t fd) {
    return TriggerEvent(fd, EPOLLIN | EPOLLOUT);
}

void IoWorker::OnTimerInsertedAtFront() {
    Tickle();
}

IoWorker::FdContext::Ptr IoWorker::GetFdContext(int fd, bool auto_create) {
    auto it = fd_contexts_.find(fd);
    if ( it == fd_contexts_.end()) {
        FdContext::Ptr fd_ctx = nullptr;
        if (auto_create) {
            fd_ctx = std::shared_ptr<FdContext>(new FdContext);
            fd_ctx->fd = fd;
            fd_contexts_[fd] = fd_ctx;
        }
        return fd_ctx;
    }
    return it->second;
}

IoManager::IoManager(size_t threads, const std::string& name)
        : Scheduler(threads, name){
    // 开始
    Start();
}

IoManager::~IoManager() {
//    OnIdle();
    Stop();
}

void IoManager::Dump(std::ostream &os) const {
    os << "IoManager{";
    Scheduler::Dump(os);
    os << "}";
}

void IoManager::Tickle(int thread_id) {
    if (thread_id != -1) {
        GetIoWokerByThreadId(thread_id)->Tickle();
    }else {
        for (auto& it : io_workers_) {
            it.second->Tickle();
        }
    }
}

void IoManager::OnIdle() {
    GetCurrentIoWorker()->OnWorking();
}

bool IoManager::IsStopped() {
    return Scheduler::IsStopped();
}

int IoManager::DelEvent(size_t fd, uint32_t evs) {
    return GetIoWorkerByDispatchId(fd)->DelEvent(fd, evs);
}

int IoManager::TriggerEvent(size_t fd, uint32_t evs) {
    return GetIoWorkerByDispatchId(fd)->TriggerEvent(fd, evs);
}

int IoManager::CancelAll(size_t fd) {
    return GetIoWorkerByDispatchId(fd)->CancelAll(fd);
}

Timer::Ptr IoManager::AddTimer(uint64_t ms, std::function<void()> cb, bool recurring) {
    return GetCurrentIoWorker()->AddTimer(ms, cb, recurring);
}

Timer::Ptr IoManager::AddConditionTimer(uint64_t ms,
                                        std::function<void()> cb,
                                        std::weak_ptr<void> cond_weak,
                                        bool recurring) {
    return GetCurrentIoWorker()->AddConditionTimer(ms, cb, cond_weak, recurring);
}

IoManager *IoManager::GetCurrentIoManager() {
    return dynamic_cast<IoManager*>(Scheduler::GetCurrentScheduler());
}

IoWorker::Ptr IoManager::GetCurrentIoWorker() {
    auto iom = dynamic_cast<IoManager*>(Scheduler::GetCurrentScheduler());
    HSBY_ASSERT(iom);
    return iom->GetIoWokerByThreadId(GetThisThreadId());
}

IoWorker::Ptr IoManager::GetIoWokerByThreadId(int thread_id) {
    auto& io_worker = io_workers_[thread_id];
    if (!io_worker) {
        io_worker = std::shared_ptr<IoWorker>(new IoWorker(this));
    }
    return io_worker;
}

IoWorker::Ptr IoManager::GetIoWorkerByDispatchId(int id) {
    return GetIoWokerByThreadId(worker_thread_ids[id % worker_thread_ids.size()]);
}

} // namespace hsby